package storm.cookbook;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

import java.util.Map;
import java.util.Random;

/**
 * Created by chenlong on 2/15/15.
 */
public class HelloWorldSpout extends BaseRichSpout {

    /**
     * Spout输出结果收集器
     */
    private SpoutOutputCollector collector;

    /**
     * 供参考的随机数
     */
    private int referenceRandom;

    private static final int MAX_RANDOM = 10;

    public HelloWorldSpout() {
        final Random rand = new Random();
        referenceRandom = rand.nextInt(MAX_RANDOM);
    }

    /**
     * Storm集群会调用open(...)方法打开Spout
     * @param map
     * @param topologyContext
     * @param spoutOutputCollector
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.collector = spoutOutputCollector;
    }

    /**
     * Storm集群会重复调用nextTuple()方法。该方法包含了整个Spout的逻辑。
     */
    @Override
    public void nextTuple() {
        Utils.sleep(100);
        final Random rand = new Random();
        // 实例随机数
        int instanceRandom = rand.nextInt(MAX_RANDOM);
        // 与参考随机数比较是否相等
        if (instanceRandom == referenceRandom) {
            collector.emit(new Values("Hello World"));
        } else {
            collector.emit(new Values("Other Random Word"));
        }
    }

    /**
     * 通过declareOutputFields(...)告诉Storm集群Spout发送了哪些字段
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("sentence"));
    }
}
